package com.atguigu.util;

import com.atguigu.common.GmallConfig;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.Properties;

public class KafkaUtil_wm {

    public static FlinkKafkaConsumer<String> getFlinkKafkaConsumer(String topic, String groupId) {

        Properties properties = new Properties();
        properties.setProperty( ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, GmallConfig.KFKA_SERVER );
        properties.setProperty( ConsumerConfig.GROUP_ID_CONFIG, groupId );

        return new FlinkKafkaConsumer<String>( topic, new KafkaDeserializationSchema<String>() {
            @Override
            public boolean isEndOfStream(String nextElement) {
                return false;
            }

            @Override
            public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
                if (record != null && record.value() != null) {
                    return new String( record.value() );
                }
                return null;
            }

            @Override
            public TypeInformation<String> getProducedType() {
                return BasicTypeInfo.STRING_TYPE_INFO;
            }
        }, properties );
    }


    // TODO: 2023/3/27 往 kafka 主题写数据
    public static <T> FlinkKafkaProducer<T> getFlinkKafkaProducer_T(KafkaSerializationSchema<T> kafkaSerializationSchema) {
        Properties properties = new Properties();
        properties.setProperty( ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, GmallConfig.KFKA_SERVER );
        return new FlinkKafkaProducer<T>( "dwd_topic_edu",
                kafkaSerializationSchema,
                properties,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }

    // TODO: 2023/3/28 从kafka主题消费数据  DDL语句
    public static String getKafkaDDL(String topic, String groupId) {
        return " with(" +
                "'connector'='kafka'," +
                "'topic'='" + topic + "'," +
                "'properties.bootstrap.servers'='" + GmallConfig.KFKA_SERVER + "'," +
                "'properties.group.id'='" + groupId + "'," +
                "'format'='json'," +
                "'scan.startup.mode'='group-offsets')";
    }

    // TODO: 2023/3/29 写入kafka 主题  DDL语句
    public static String getKafkaSinkDDL(String topic) {
        return "with(" +
                "'connector'='kafka'," +
                "'topic'='" + topic + "'," +
                "'properties.bootstrap.servers'='" + GmallConfig.KFKA_SERVER + "'," +
                "'format'='json'  " +
                ")";
    }
}